package cn.anjubao.cp.api.config.mq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;


@Configuration
public class RocketMQConfig {

    private static final Logger LOG = LoggerFactory.getLogger(RocketMQConfig.class);

    @Autowired
    private RocketMQProperties rocketMQProperties;

    @Autowired
    private ThreadPoolTaskExecutor executor;

    @Autowired
    private MessageHandler messageHandler;

    /**
     * MQ 是基于发布订阅模型的消息系统。在 MQ 消息系统中消息的订阅方订阅关注的 Topic，以获取并消费消息。由于订阅方应用一般是分布式系统，以集群方式部署有多台机器。因此 MQ 约定以下概念。
     *
     * 集群：MQ 约定使用相同 Consumer ID 的订阅者属于同一个集群，同一个集群下的订阅者消费逻辑必须完全一致（包括 Tag 的使用），这些订阅者在逻辑上可以认为是一个消费节点。
     *
     * 集群消费：当使用集群消费模式时，MQ 认为任意一条消息只需要被集群内的任意一个消费者处理即可。
     *
     * 广播消费：当使用广播消费模式时，MQ 会将每条消息推送给集群内所有注册过的客户端，保证消息至少被每台机器消费一次。
     * @return
     * @throws Exception
     */
    @Bean
    public DefaultMQPushConsumer consumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(rocketMQProperties.getConsumer().getGroup());
        consumer.setNamesrvAddr(rocketMQProperties.getNamesrv());
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.subscribe(rocketMQProperties.getConsumer().getTopic(), "*");
        consumer.registerMessageListener(messageHandler);
        consumer.start();
        return consumer;
    }

    @Bean
    public DefaultMQProducer producer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer(rocketMQProperties.getProducer().getGroup());
        producer.setNamesrvAddr(rocketMQProperties.getNamesrv());
        producer.start();
        return producer;
    }
}
